package com.you.raft.core.impl;

import com.sun.org.apache.xerces.internal.impl.xs.opti.DefaultNode;
import com.you.raft.common.entity.*;
import com.you.raft.core.Consensus;
import com.you.raft.core.LogModule;
import com.you.raft.core.StateMachine;
import io.netty.util.internal.StringUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.locks.ReentrantLock;

/**
 * 一致性模块
 */
@Slf4j
public class DefaultConsensus implements Consensus {

    private DefaultRaftNode node;
    private StateMachine stateMachine;
    private LogModule logModule;
    private Cluster cluster;

    private final ReentrantLock voteLock = new ReentrantLock();
    private final ReentrantLock appendLock = new ReentrantLock();

    public DefaultConsensus(DefaultRaftNode node) {
        this.node = node;
        this.cluster = node.getCluster();
        this.stateMachine = node.getStateMachine();
        this.logModule = node.getLogModule();
    }

    @Override
    public VoteResponse requestVote(VoteRequest request) {
        try {
            if (!voteLock.tryLock()) {
                log.warn("拒绝投票:当前有其他节点正在处理投票请求！");
                return VoteResponse.builder().term(node.getCurrentTerm()).voteGranted(false).build();
            }
            if (request.getTerm() < node.getCurrentTerm()) {
                log.warn("拒绝投票:请求投票的任期小于当前节点的任期!");
                return VoteResponse.builder().term(node.getCurrentTerm()).voteGranted(false).build();
            }

            // (当前节点并没有投票 或者 已经投票过了且是对方节点) && 对方日志和自己一样新
            log.info("节点已经投票给: [{}], 请求投票的节点:[{}]", node.getVotedFor(), request.getCandidateId());
            log.info("当前节点任期: {}, 请求投票的节点任期: {}", node.getCurrentTerm(), request.getTerm());

            if ((StringUtil.isNullOrEmpty(node.getVotedFor()) || node.getVotedFor().equals(request.getCandidateId()))) {

                if (node.getLogModule().getLast() != null) {
                    // 对方没有自己新
                    if (node.getLogModule().getLast().getTerm() > request.getLastLogTerm()) {
                        return VoteResponse.fail();
                    }
                    // 对方没有自己新
                    if (node.getLogModule().getLastIndex() > request.getLastLogIndex()) {
                        return VoteResponse.fail();
                    }
                }
                // 切换状态
                node.setStatus(NodeStatus.FOLLOWER);
                // 更新
                cluster.setLeader(new Node(request.getCandidateId()));
                node.setCurrentTerm(request.getTerm());
                node.setVotedFor(request.getCandidateId());
                log.warn("成功投票给:{}", request.getCandidateId());
                // 返回成功
                return VoteResponse.builder().term(node.getCurrentTerm()).voteGranted(true).build();
            }
            log.warn("拒绝投票:当前节点已经给别的节点投票了!");
            return VoteResponse.builder().term(node.getCurrentTerm()).voteGranted(false).build();
        } finally {
            voteLock.unlock();
        }
    }

    @Override
    public AppendEntryResponse appendEntry(AppendEntryRequest request) {
        AppendEntryResponse response = AppendEntryResponse.fail();
        try {
            if (!appendLock.tryLock()) {
                return response;
            }
            // 任期校验
            response.setTerm(node.getCurrentTerm());
            if (request.getTerm() < node.getCurrentTerm()) {
                return response;
            }
            node.setPreHeartBeatTime(System.currentTimeMillis());
            node.setPreElectionTime(System.currentTimeMillis());
            cluster.setLeader(new Node(request.getLeaderId()));
            node.setStatus(NodeStatus.FOLLOWER);
            node.setCurrentTerm(request.getTerm());
            //心跳判断-处理LEADER节点已提交但未应用到状态机的日志
            //如果 leaderCommit > commitIndex，令 commitIndex 等于 leaderCommit 和 新日志条目索引值中较小的一个
            if (request.getEntries() == null || request.getEntries().length == 0) {
                log.info("收到领导者 {} 的心跳,任期: {}, 当前节点的任期: {}",
                        request.getLeaderId(), request.getTerm(), node.getCurrentTerm());
                long nextCommit = node.getCommitIndex() + 1;
                if (request.getLeaderCommit() > node.getCommitIndex()) {
                    int commitIndex = (int) Math.min(request.getLeaderCommit(), node.getLogModule().getLastIndex());
                    node.setCommitIndex(commitIndex);
                    node.setLastApplied(commitIndex);
                }
                while (nextCommit <= node.getCommitIndex()) {
                    // 应用提交的日志
                    stateMachine.apply(logModule.get(nextCommit));
                    nextCommit++;
                }
                return AppendEntryResponse.builder().term(node.getCurrentTerm()).success(true).build();
            }

            // 在当前日志查找是否是否存在一个日志是否与请求前一个日志的任期号与日志号完全相同
            // 如果不能找到，拒绝请求
            // 能找到，擦除该日志(该日志任期号+1）后面的所有日志,然后插入请求的日志
            LogEntry logEntry;
            if ((logEntry = node.getLogModule().get(request.getPrevLogIndex())) != null) {
                if (logEntry.getTerm() != request.getPreLogTerm()) {
                    return response;
                }
            } else {
                return response;
            }
            logModule.removeStartWith(logEntry.getIndex() + 1);
            for (LogEntry entry : request.getEntries()) {
                node.getLogModule().write(entry);
                response.setSuccess(true);
            }
            // 下一个需要提交的日志的索引（如有）
            //如果 leaderCommit > commitIndex，令 commitIndex 等于 leaderCommit 和 新日志条目索引值中较小的一个
            long nextCommit = node.getCommitIndex() + 1;
            if (request.getLeaderCommit() > node.getCommitIndex()) {
                int commitIndex = (int) Math.min(request.getLeaderCommit(), node.getLogModule().getLastIndex());
                node.setCommitIndex(commitIndex);
                node.setLastApplied(commitIndex);
            }
            while (nextCommit <= node.getCommitIndex()) {
                // 提交之前的日志
                stateMachine.apply(logModule.get(nextCommit));
                nextCommit++;
            }
            return response;
        } finally {
            appendLock.unlock();
        }
    }
}
